package git.soulbgm.common;

import git.soulbgm.log.LogOutput;
import git.soulbgm.thread.ThreadTool;
import git.soulbgm.utils.JsonTool;
import git.soulbgm.utils.LogHelper;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

/**
 * kafka操作类
 *
 * @author SoulBGM
 * @date 2018-06-14 17:16
 */
@Component
public class KafkaOperate {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    public KafkaOperate(KafkaTemplate<String, Object> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    private static final LogHelper LOG = new LogHelper(KafkaOperate.class);

    /**
     * 将数据存入kafka
     *
     * @param topicName 主题名称
     * @param data      数据
     * @return {@link ListenableFuture}
     */
    public ListenableFuture<SendResult<String, Object>> dataCache(final String topicName, final Object data) {
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topicName, data);
        future.addCallback(
                result -> LogOutput.kafkaSuccessOutputLog(topicName, data),
                ex -> LOG.error("Topic: {},存入Kafka失败,数据: {},异常: ", topicName, JsonTool.toJson(data), ex)
        );
        return future;
    }

}
